package cn.itcast.map;

import cn.itcast.bean.StockBean;
import cn.itcast.config.QuotConfig;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import org.apache.commons.lang.text.StrBuilder;
import org.apache.flink.api.common.functions.RichMapFunction;

import java.sql.Timestamp;

/**
 * 将个股分时数据,转成字符串,写入hdfs,后续再插入hive
 */
public class StockPutHdfsMap extends RichMapFunction<StockBean,String> {
    //1.定义字符串字段分隔符
    String sp = QuotConfig.config.getProperty("hdfs.seperator");

    @Override
    public String map(StockBean value) throws Exception {

        /**
         * 把bean对象转换成字符串 -> 方便加载数据到hive数仓
         * 字符串拼装字段顺序：
         * Timestamp|date|secCode|secName|preClosePrice|openPirce|highPrice|
         * lowPrice|closePrice|tradeVol|tradeAmt|tradeVolDay|tradeAmtDay|source
         */
        StringBuffer sb = new StringBuffer();
            sb.append(value.getEventTime()).append(sp)
                    .append(value.getTradeDate()) .append(sp)
                    .append(value.getSecCode()).append(sp)
                    .append(value.getSecName()).append(sp)
                    .append(value.getPreClosePrice()).append(sp)
                    .append(value.getOpenPrice()).append(sp)
                    .append(value.getHighPrice()).append(sp)
                    .append(value.getLowPrice()).append(sp)
                    .append(value.getClosePrice()).append(sp)
                    .append(value.getTradeVol()).append(sp)
                    .append(value.getTradeAmt()).append(sp)
                    .append(value.getTradeVolDay()).append(sp)
                    .append(value.getTradeAmtDay()).append(sp)
                    .append(value.getSource());


        return sb.toString();
    }
}
